package com.framework.rocketmq.consumer;

import cn.hutool.core.util.IdUtil;
import com.framework.rocketmq.config.RocketMQConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;

import javax.annotation.PostConstruct;
import java.util.List;

/**
 * 描述：
 * 消费者
 * @author xianggj
 * @Date 2021/10/13 14:35
 **/
@Slf4j
public abstract class Consumer {

    @Autowired
    private RocketMQConfig rocketMQConfig;

    /**
     * 特殊情况下可以覆盖原有配置
     * @param rocketMQConfig
     */
    protected void setRocketMQConfig(RocketMQConfig rocketMQConfig) {
        this.rocketMQConfig = rocketMQConfig;
    }

    private DefaultMQPushConsumer consumer;

    /**
     * 初始化
     * @return
     */
    @PostConstruct
    public void init() throws MQClientException {
        MQConfig mqCfg = getMQConfig();
        consumer = new DefaultMQPushConsumer();
        consumer.setConsumerGroup(getGroupName());
        consumer.setNamesrvAddr(rocketMQConfig.getNameSrvAddr());
        consumer.setConsumeFromWhere(mqCfg.getConsumeFromWhere());
        //如果配置 ConsumeFromWhere.CONSUME_FROM_TIMESTAMP 需要配置这里
//        consumer.setConsumeTimestamp(UtilAll.timeMillisToHumanString());
        consumer.setMessageModel(mqCfg.getMessageModel());

        if (StringUtils.isNotBlank(rocketMQConfig.getInstanceName())){
            consumer.setInstanceName(rocketMQConfig.getInstanceName());
        } else {
            consumer.setInstanceName(IdUtil.fastSimpleUUID());
        }

        consumer.setConsumeThreadMax(mqCfg.getMaxThread());
        consumer.setConsumeThreadMin(mqCfg.getMinThread());



        consumer.subscribe(mqCfg.getTopic(), mqCfg.getSubExpression());


        // 这里可以抽离出来，添加一个继承MessageListenerConcurrently的类
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                if (null != msgs && !msgs.isEmpty()) {
                    for (MessageExt msg : msgs) {
                        if (msg.getReconsumeTimes() > 0){
                            continue;
                        }
                        if (mqCfg.getTopic().equals(msg.getTopic())) {

                            try {
                                consume(msg, context);
                            } catch (Exception e) {
                                log.error(e.getMessage(), e);
                                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                            }
                        } else {
                            // 如果没有return success, consumer会重新消费该消息, 直到return success
                            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                        }
                    }
                    // 如果没有return success, consumer会重新消费该消息, 直到return success
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // Consumer对象在使用之前必须要调用start初始化, 初始化一次即可
        consumer.start();
        log.info("ConsumerGroup: {}, Topic: {}, SubExpression: {} started!",
                rocketMQConfig.getConsumerGroup(),
                mqCfg.getTopic(), mqCfg.getSubExpression());
    }


    protected abstract void consume(MessageExt msg, ConsumeConcurrentlyContext context) throws Exception;

    protected abstract MQConfig getMQConfig();

    /**
     * 一个实例一个组
     * @return
     */
    protected abstract String getGroupName();

}
